package com.zcy;

import com.espertech.esper.common.client.EPCompiled;
import com.espertech.esper.common.client.EventBean;
import com.espertech.esper.common.client.configuration.Configuration;
import com.espertech.esper.compiler.client.CompilerArguments;
import com.espertech.esper.compiler.client.EPCompileException;
import com.espertech.esper.compiler.client.EPCompiler;
import com.espertech.esper.compiler.client.EPCompilerProvider;
import com.espertech.esper.runtime.client.*;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

/**
 * 对于Schema来说没有插入流的概念
 */
@Slf4j
public class NonSelectAddListener {

    private static final String _SuperSchema = "create schema SecurityEvent as (ipAddress string, userId String, numAttempts int);";
    // copy from 只会复制 SuperSchema的属性，但不会要求两边的表达形式一致（objectarray、json等）
    // 但是 关键字 inherits 会要求两边格式对等
    // @buseventtype 需要配合 @public 一起使用
    private static final String _CreatePersonSchema = "@public @buseventtype @name('non-select') create schema SimilarSecurityEvent () copyfrom SecurityEvent;";

    @Test
    public void createSchemaTest() throws InterruptedException, EPCompileException, EPDeployException {

        EPCompiler compiler = EPCompilerProvider.getCompiler();
        Configuration configuration = new Configuration();

        configuration.getCommon().addMapSuperType("SimilarSecurityEvent", "SecurityEvent");

        CompilerArguments compilerArguments = new CompilerArguments(configuration);

        EPCompiled compile = compiler.compile(_SuperSchema + "\n" + _CreatePersonSchema, compilerArguments);

        EPRuntime runtime = EPRuntimeProvider.getDefaultRuntime();
        EPDeploymentService deploymentService = runtime.getDeploymentService();

        EPDeployment deploy = deploymentService.deploy(compile);
        EPStatement statement = deploymentService.getStatement(deploy.getDeploymentId(), "non-select");
        statement.addListener(new UpdateListener() {
            @Override
            public void update(EventBean[] newEvents, EventBean[] oldEvents, EPStatement statement, EPRuntime runtime) {
                Arrays.stream(newEvents).forEach(item -> {
                    Object underlying = item.getUnderlying();
                    log.info("{}", underlying);
                });
            }
        });


        EPEventService eventService = runtime.getEventService();

        Map<String, Object> map = new HashMap<>();
        map.put("ipAddress", "");
        map.put("userId", "");
        map.put("numAttempts", 1);
        eventService.sendEventMap(map, "SimilarSecurityEvent");

        Thread.sleep(1000000);
    }
}
